use crate::core::config::Conf;
use serde_derive::{Serialize, Deserialize};
use serde_json::{Map,Value};
use bytes::BytesMut;
use kafka::producer::{DefaultPartitioner, Producer, Record, RequiredAcks};
use std::thread;
use crossbeam_channel::{Receiver,Sender};
//use std::sync::mpsc::{Receiver,Sender};
use std::collections::HashMap;
use log::{trace, error};

#[derive(Debug, Serialize, Deserialize)]
struct Pack {
    fun : String,
    s   : String,
    t   : String,
    p   : String,
    k   : String,
    v   : Map<String, Value>
}


pub fn init_producer_group(thread_num : usize,stat_sender : Sender<(String,u64)>, channels : Vec<Receiver<BytesMut>>){
    let c = Conf::get_config(None);
    let mut handles = vec![];
    let mut chns = channels;
    for i in 0..thread_num {
        let stat_count = c.stat_count;
        let rx = match chns.pop() {
            Some(_rx) => _rx,
                   _  => {
                       error!("init producer group channel error for num {}, ~",i);
                       std::process::exit(0);
                   }
        };
        let status_sender = stat_sender.clone();
        let brokers = c.brokers.clone();
        let mut pdc = match Producer::from_hosts(brokers)
        .with_ack_timeout(std::time::Duration::from_secs(c.timeout))
        .with_required_acks(RequiredAcks::One)
        .create() {
            Ok(_pdc) => _pdc,
                   _  => {
                       error!("init pdc error for num {}, ~",i);
                       std::process::exit(0);
                   }
        };
        let handle = thread::spawn(move || {
            let mut stat_data : HashMap<String,u64> = HashMap::new();
            loop {
                match rx.recv() {
                    Ok(m) => {
                        let p : Pack = match serde_json::from_slice(&m[..]) {
                            Ok(r) => r,
                                _ => {
                                    trace!("kafka1 err msg is : {:?}", std::str::from_utf8(&m[..]));
                                    continue;
                                },
                        };
                        let topic = p.t.as_str();
                        trace!("find the topic : {} , {}",topic,pdc.client().topics().contains(topic));
                        if !pdc.client().topics().contains(topic) {
                            match pdc.load_topics(&[topic],DefaultPartitioner::default()) {
                                Ok(_) => {
                                    trace!("form {} load the new topic is : {:?}", i, topic);
                                },
                                _ => {
                                    trace!("failed to load the new topic is : {:?} -- {}", topic,i);
                                    continue;
                                },
                            };
                        }
                        let data = match serde_json::to_string(&p.v) {
                            Ok(_data) => _data,
                                   _  => {
                                    trace!("kafka2 err msg is : {:?}", std::str::from_utf8(&m[..]));
                                    continue;
                                },
                        };
                        match pdc.send(&Record {
                            topic: topic,
                            partition: -1,
                            key: (),
                            value: data,
                        }) {
                            Ok(_) => {
                                //status data collect
                                if stat_data.contains_key(topic) {
                                    stat_data.insert(topic.to_owned(),stat_data[topic]+1);
                                    if stat_data[topic] > stat_count {
                                        match status_sender.send((topic.to_owned(),stat_data[topic])) {
                                            _ => {
                                                stat_data.insert(topic.to_owned(),0);
                                            }
                                        };
                                    }
                                }else{
                                    stat_data.insert(topic.to_owned(),1);
                                }
                            }
                            Err(e) => {
                                trace!("kafka < {} > send err : {:?}", i, e);
                                continue;
                            },
                        }
                    },
                    Err(e)  => {
                        trace!("kafka recv : {:?}", e);
                        continue;
                    },
                }
                
            }
        });
        handles.push(handle);
    }
}
